package com.chenjj.bigdata.flink.book

import com.chenjj.bigdata.flink.watermark.{PeriodicAssigner, PunctuatedAssigner}
import org.apache.flink.api.common.ExecutionConfig
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment, _}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.junit.{Assert, Test}


/**
  * 第4.1章：时间概念与watermark
  */
@Test
class FlinkTesterChapter4_2 extends  Assert{
  val env = StreamExecutionEnvironment.getExecutionEnvironment

  /**
    * Watermarks：通过定义SourceFunction设定Timestamps和Watermarks
    */
  @Test
  def test1:Unit={
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val input = List(("a",1000L,1),("b",1L,1),("c",3L,1),("a",4000L,1),("a",20000L,1),("a",25000L,1))
    val source = env.addSource(new SourceFunction[(String,Long,Int)] {
      override def run(ctx: SourceFunction.SourceContext[(String, Long, Int)]): Unit = {
        input.foreach(element=>{
          //给每个元素打上event time标签
          ctx.collectWithTimestamp(element,element._2)
          //指定watermark的生成策略，设定元素最迟到达的延迟时间：1秒
          ctx.emitWatermark(new Watermark(element._2 -1000))
        })
      }

      override def cancel(): Unit = ???
    })

    val result = source.keyBy(0).timeWindow(Time.seconds(1)).sum("_3")

    result.print()
    env.execute()
  }

  /**
    * Watermarks(周期性水印生成)：使用Timestamp Assigner(Ascending Timestamp Assigner)指定Timestamps和Watermarks
    */
  @Test
  def test2:Unit={

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val data = List(("a",1L,1),("b",1L,1),("c",3L,1),("a",4L,1),("a",20000L,1),("a",25000L,1))
    val input = env.fromCollection(data)

    //使用系统默认Ascending分配时间信息和watermark
    val withTimestampsAndWatermarks = input.assignAscendingTimestamps(t=>t._2)
    //对数据集进行计算
    val result = withTimestampsAndWatermarks.keyBy(0).timeWindow(Time.seconds(10)).sum("_3")

    result.print()

    env.execute()

    //输出，说明：("a",20000L,1),("a",25000L,1) 和 其他的a的时间相差20秒，没有落在同一个窗口中
    //3> (b,1,1)
    //6> (c,3,1)
    //8> (a,1,2)
    //8> (a,20000,2)
  }

  /**
    * Watermarks(周期性水印生成)：使用Timestamp Assigner(固定时延间隔)指定Timestamps和Watermarks
    */
  @Test
  def test3:Unit={

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val data = List(("a",1L,1),("b",1L,1),("c",3L,1),("a",4L,1),("a",20000L,1),("a",25000L,1))
    val input = env.fromCollection(data)
    //使用BoundedOutOfOrdernessTimestampExtractor指定timestamp
    val withTimestampsAndWatermarks = input.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Int)](Time.seconds(10)) {
      //定义抽取EventTime Timestamp逻辑
      override def extractTimestamp(element: (String, Long, Int)): Long = {
        element._2
      }
    })
    //对数据集进行计算
    val result = withTimestampsAndWatermarks.keyBy(0).timeWindow(Time.seconds(10)).sum("_3")

    result.print()

    env.execute()

    //输出，说明：("a",20000L,1),("a",25000L,1) 和 其他的a的时间相差20秒，没有落在同一个窗口中
    //3> (b,1,1)
    //6> (c,3,1)
    //8> (a,1,2)
    //8> (a,20000,2)
  }


  /**
    * 自定义Timestamp Assigner和Watermark Generator : Periodic Watermarks
    */
  def test4:Unit={
      val a1 = new PeriodicAssigner
      val a2 = new PunctuatedAssigner
  }
}
